package org.zhaya.sdk.service;

import cn.hutool.core.lang.Assert;
import cn.hutool.core.lang.ClassScanner;
import cn.hutool.core.lang.Pair;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.ClassUtil;
import com.google.common.collect.ImmutableList;
import io.logz.sawmill.annotations.ProcessorProvider;
import io.logz.sawmill.processors.AddTagProcessor;
import org.zhaya.sdk.DodpProcessor;

import java.util.List;
import java.util.Map;
import java.util.ServiceLoader;
import java.util.Set;

/**
 * @Author Ryan Yang
 * @Date 2022/10/10 9:59 上午
 */
public class ProcessorManager {

    private static final List<String> PROCESSOR_PACKAGE_SET = ImmutableList.<String>builder()
            .add(ClassUtil.getPackage(AddTagProcessor.class))
            .add(ClassUtil.getPackage(DodpProcessor.class))
            .build();

    /**
     * key className, value pair.key classSimpleName  pair.value type
     */
    private static final Map<String, Pair<String, String>> PROCESSOR_CLASS_REF_TYPE_MAP = MapUtil.newConcurrentHashMap(64);
    /**
     * key type, value className
     */
    private static final Map<String, String> TYPE_REF_PROCESSOR_CLASS_MAP = MapUtil.newConcurrentHashMap(64);

    public static final String IF_PROCESSOR_TYPE = "if";
    public static final String DEBUG_PROCESSOR_TYPE = "debug";
    public static final String KAFKA_SOURCE_PROCESSOR_TYPE = "kafka";
    public static final String KAFKA_SINK_PROCESSOR_TYPE = "kafkaSink";
    public static final String DATA_STORE_SINK_PROCESSOR_TYPE = "kafkaSinkProcessor";

    /**
     * 扫描组件信息
     */
    static {
        PROCESSOR_PACKAGE_SET.forEach(processorPackage -> {
            Set<Class<?>> classSet = ClassScanner.scanPackageByAnnotation(processorPackage, ProcessorProvider.class);
            classSet.forEach(clazz -> {
                PROCESSOR_CLASS_REF_TYPE_MAP.put(clazz.getName(), Pair.of(clazz.getSimpleName(), clazz.getAnnotation(ProcessorProvider.class).type()));
                TYPE_REF_PROCESSOR_CLASS_MAP.put(clazz.getAnnotation(ProcessorProvider.class).type(), clazz.getName());
            });
        });
        ServiceLoader<DodpProcessor> spiProcessors = ServiceLoader.load(DodpProcessor.class);
        spiProcessors.forEach(spiProcessor -> {
            Class<?> spiProcessorClass = spiProcessor.getClass();
            PROCESSOR_CLASS_REF_TYPE_MAP.put(spiProcessorClass.getName(), Pair.of(spiProcessorClass.getSimpleName(), spiProcessorClass.getAnnotation(ProcessorProvider.class).type()));
            TYPE_REF_PROCESSOR_CLASS_MAP.put(spiProcessorClass.getAnnotation(ProcessorProvider.class).type(), spiProcessorClass.getName());
        });
        TYPE_REF_PROCESSOR_CLASS_MAP.put(IF_PROCESSOR_TYPE, IF_PROCESSOR_TYPE);
        TYPE_REF_PROCESSOR_CLASS_MAP.put(KAFKA_SOURCE_PROCESSOR_TYPE, KAFKA_SOURCE_PROCESSOR_TYPE);
    }

    public static Pair<String, String> getTypeInfo(String processorClassName) {
        Assert.notBlank(processorClassName, "processor class is blank");
        return PROCESSOR_CLASS_REF_TYPE_MAP.get(processorClassName);
    }

    public static boolean checkTypeExist(String processorType) {
        Assert.notBlank(processorType, "processor type is blank");
        return TYPE_REF_PROCESSOR_CLASS_MAP.containsKey(processorType);
    }

    public static boolean checkIsIf(String processorType) {
        Assert.notBlank(processorType, "processor type is blank");
        return IF_PROCESSOR_TYPE.equalsIgnoreCase(processorType);
    }

    public static boolean checkIsDebug(String processorType) {
        Assert.notBlank(processorType, "processor type is blank");
        return DEBUG_PROCESSOR_TYPE.equalsIgnoreCase(processorType);
    }

    public static boolean checkIsSource(String processorType) {
        Assert.notBlank(processorType, "processor type is blank");
        return KAFKA_SOURCE_PROCESSOR_TYPE.equalsIgnoreCase(processorType);
    }

    public static boolean checkIsSink(String processorType) {
        Assert.notBlank(processorType, "processor type is blank");
        return KAFKA_SINK_PROCESSOR_TYPE.equalsIgnoreCase(processorType) || DATA_STORE_SINK_PROCESSOR_TYPE.equalsIgnoreCase(processorType);
    }

    public static boolean checkIsSourceOrSink(String processorType) {
        Assert.notBlank(processorType, "processor type is blank");
        return checkIsSource(processorType) || checkIsSink(processorType);
    }

}
